package org.qiyu.live.id.generate.privoder.service.impl;


import jakarta.annotation.Resource;
import org.qiyu.live.id.generate.privoder.dao.mapper.IdGenerateMapper;
import org.qiyu.live.id.generate.privoder.dao.po.IdGeneratePO;
import org.qiyu.live.id.generate.privoder.service.IdGenerateService;
import org.qiyu.live.id.generate.privoder.service.bo.LocalSeqIdBO;
import org.qiyu.live.id.generate.privoder.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;


import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(IdGenerateServiceImpl.class);

    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap=new ConcurrentHashMap<>();
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap=new ConcurrentHashMap<>();

    private static final Float UPDATE_RATE=0.75f;

    private static final int SEQ_ID = 1;

    private static final Map<Integer, Semaphore> semaphoreMap  =new ConcurrentHashMap<>();

    private static final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.MINUTES, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("id-generate-thread-"+ThreadLocalRandom.current().nextInt(1000));
            return thread;
        }
    });

    @Resource
    private IdGenerateMapper idGenerateMapper;

    @Override
    public Long getSeqId(Integer id) {
        //1 校验参数
        if(id == null){
            LOGGER.error("[getSeqId] id is error,id is {}",id);
        }
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);
        //校验localSeqIdBO是否为空
        if(localSeqIdBO == null){
            LOGGER.error("[getSeqId] id is error,id is {}",id);
        }
        //当我们map中的数字数量用到75%的时候去触发更新map的操作
        this.refreshLocalSeqId(localSeqIdBO);
        long returnId = localSeqIdBO.getCurrentNum().getAndIncrement();
        if(returnId>localSeqIdBO.getNextThreshold()){
            LOGGER.error("[getSeqId] id is over limit,id is {}",id);
            return null;
        }
        return returnId;
    }

    @Override
    public Long getUnSeqId(Integer id) {
        //校验参数
        if(id== null){
            LOGGER.error("[getUnSeqId] id is error,id is {}",id);
        }
        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(id);
        if(localUnSeqIdBO == null){
            LOGGER.error("[getUnSeqId] id is error,id is {}",id);
        }
        Long returnId = localUnSeqIdBO.getIdqueue().poll();
        if(returnId>localUnSeqIdBO.getNextThreshold()){
            LOGGER.error("[getSeqId] id is over limit,id is {}",id);
            return null;
        }
        this.refreshLocalUnSeqId(localUnSeqIdBO);
        return returnId;
    }
    //当我们map中的数字数量用到75%的时候去触发更新map的操作
    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        Long begin = localUnSeqIdBO.getCurrentStart();
        Long end = localUnSeqIdBO.getNextThreshold();
        int remainSize = localUnSeqIdBO.getIdqueue().size();
        //当用到75%的时候触发更新map的操作
        if((end-begin)*0.25>remainSize){
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if(semaphore == null){
                LOGGER.error("semaphore is null,id is {}", localUnSeqIdBO.getId());
            }
            boolean tryAcquire = semaphore.tryAcquire();
            if(tryAcquire){
                LOGGER.info("尝试进行无序id同步");
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                            tryUpdateMySQLRecord(idGeneratePO);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        } finally {
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                        }
                    }
                });
            }
        }
    }




    //当我们map中的数字数量用到75%的时候去触发更新map的操作
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        Long nextThreshold = localSeqIdBO.getNextThreshold();
        Long currentStart = localSeqIdBO.getCurrentStart();
        Long step = nextThreshold-currentStart;
        //当用到75%的时候触发更新map的操作
        if((localSeqIdBO.getCurrentNum().get()-currentStart)>UPDATE_RATE*step){
            //获取信号量，只允许一个线程去更新map中的数据
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());

            if(semaphore == null) {
                LOGGER.error("semaphore is null,id is {}", localSeqIdBO.getId());
            }

            boolean tryAcquire = semaphore.tryAcquire();
            if(tryAcquire){
                //另起一个线程去更新map中的数据
                LOGGER.info("尝试进行id同步");
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                            tryUpdateMySQLRecord(idGeneratePO);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        } finally {
                            //释放信号量
                            semaphoreMap.get(localSeqIdBO.getId()).release();
                        }
                    }
                });
            }

        }
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        //去初始化我们的localSeqIdBOMap
        List<IdGeneratePO> idGeneratePOS = idGenerateMapper.selectAll();
        for (IdGeneratePO idGeneratePO : idGeneratePOS) {
            tryUpdateMySQLRecord(idGeneratePO);
            semaphoreMap.put(idGeneratePO.getId(),new Semaphore(1));
        }

    }

    private void tryUpdateMySQLRecord(IdGeneratePO idGeneratePO){
        Integer updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
        //1.0如果更新成功
        if(updateResult > 0){
            localIdHandler(idGeneratePO);
            return;
        }
        //1.0如果更新失败，进行重试机制
        for (int i = 0; i < 3; i++) {
            idGeneratePO = idGenerateMapper.selectById(idGeneratePO.getId());
            updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(),idGeneratePO.getVersion());
            if(updateResult > 0){
                localIdHandler(idGeneratePO);
                return;
            }
        }
        throw new RuntimeException("表id段占用失败，竞争过于激烈，id is {}"+idGeneratePO.getId());
    }




    /**
     *专门处理将本地id对象放入到map中，并进行初始化的
     * @param idGeneratePO
     */
    private void localIdHandler(IdGeneratePO idGeneratePO){
        long currentStart = idGeneratePO.getCurrentStart();
        long nextThreshold = idGeneratePO.getNextThreshold();
        long currentNum = currentStart;
        if(idGeneratePO.getIsSeq() == SEQ_ID){
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            localSeqIdBO.setId(idGeneratePO.getId());
            AtomicLong atomicLong = new AtomicLong(currentNum);
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);
            localSeqIdBOMap.put(localSeqIdBO.getId(),localSeqIdBO);
        }else {
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGeneratePO.getId());
            localUnSeqIdBO.setCurrentStart(idGeneratePO.getCurrentStart());
            localUnSeqIdBO.setNextThreshold(idGeneratePO.getNextThreshold());
            long begin = idGeneratePO.getCurrentStart();
            long end = idGeneratePO.getNextThreshold();
            ArrayList<Long> idList = new ArrayList<>();
            for (long i = begin; i < end; i++) {
                idList.add(i);
            }
            Collections.shuffle(idList);
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>();
            idQueue.addAll(idList);
            localUnSeqIdBO.setIdqueue(idQueue);
            localUnSeqIdBOMap.put(localUnSeqIdBO.getId(),localUnSeqIdBO);
        }
    }



}
